package com.lcache.common.mq;


import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;


/**
 * @program: LIM
 * @description: rocketmq发送类
 * @author: Alex Wu
 * @createDate: 2025-04-15 14:17
 **/

@Slf4j
public class RocketMQSender {


    private static volatile RocketMQSender instance;

    private final Producer producer;
    private final ClientServiceProvider provider;
    private final String topic;

    private RocketMQSender(String endpoint, String topic) throws ClientException {
        this.topic = topic;
        this.provider = ClientServiceProvider.loadService();

        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);
        ClientConfiguration configuration = builder.build();

        this.producer = provider.newProducerBuilder()
                .setTopics(topic)
                .setClientConfiguration(configuration)
                .build();

        log.info("RocketMQ Producer initialized for topic: {}", topic);
    }

    public static RocketMQSender getInstance(String endpoint, String topic) throws ClientException {
        if (instance == null) {
            synchronized (RocketMQSender.class) {
                if (instance == null) {
                    instance = new RocketMQSender(endpoint, topic);
                }
            }
        }
        return instance;
    }

    public void send(String tag, String key, String body) {
        Message message = provider.newMessageBuilder()
                .setTopic(topic)
                .setTag(tag)
                .setKeys(key)
                .setBody(body.getBytes())
                .build();
        try {
            SendReceipt receipt = producer.send(message);
            log.info("Message sent successfully, messageId={}", receipt.getMessageId());
        } catch (ClientException e) {
            log.error("Failed to send message to topic: {}", topic, e);
        }
    }

    public void shutdown() {
        if (producer != null) {
            try {
                producer.close();
            }catch (Exception e) {
                log.error("Failed to close producer", e);
            }
        }
    }
}
